package com.example.kafkademo.multithread;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by zyy on 2021/12/29.
 */
public class KafkaConsumerRunner implements Runnable {

    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final KafkaConsumer consumer;

    public KafkaConsumerRunner(KafkaConsumer consumer) {
        this.consumer = consumer;
    }


    public void run() {
        try {
            consumer.subscribe(Arrays.asList("test-a"));
            while (!closed.get()) {
                ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));
                System.out.println("Thread name:" + Thread.currentThread().getName() + "获取到的消息数量：" + records.count());
                //  执行消息处理逻辑
            }
        } catch (WakeupException e) {
            // Ignore exception if closing
            if (!closed.get()) throw e;
        } finally {
            consumer.close();
        }
    }


    // Shutdown hook which can be called from a separate thread
    public void shutdown(KafkaConsumer consumer) {
        closed.set(true);
        consumer.wakeup();
    }
}

class buildConsumer{

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.42.25.12:9092,10.42.25.13:9092,10.42.25.15:9092");
        props.put("group.id", "12334");
        props.put("enable.auto.commit", "true");
//        props.put("auto.commit.interval.ms", "2000");     //表明 Kafka 每 5 秒会为你自动提交一次位移(默认)
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 5000); // 默认5分钟
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        for(int i = 0; i <= 2; i++){
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            Thread thread = new Thread(new KafkaConsumerRunner(consumer));
            thread.setName("new Thread======" + i +5 );
            thread.start();
        }
    }
}